package yyn

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"fyne.io/fyne/v2"
	"fyne.io/fyne/v2/container"
	"fyne.io/fyne/v2/data/binding"
	"fyne.io/fyne/v2/dialog"
	"fyne.io/fyne/v2/widget"
	"github.com/segmentio/kafka-go"
	"io"
	"log"
	"math"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	time "time"
)

//地址输入框
var addrBind = binding.NewString()

//连接信息框
var infoBind = binding.NewString()

//左侧列表数据
var topicsData = binding.BindStringList(
	&[]string{"未连接"},
)

//列表选中数据 -1表示未选中
var topicSelected = -1

var msgWait = "未连接"
var msgSuccess = "连接成功"
var msgFail = "连接失败"

//起止时间
var startBind = binding.NewString()
var endBind = binding.NewString()

//是否带时间戳
var dataTime = false

//步骤
var oneBind = binding.NewString()
var twoBind = binding.NewString()
var threeBind = binding.NewString()
var fourBind = binding.NewFloat()
var fiveBind = binding.NewString()

//日期格式验证
var dateReg = "((([0-9]{3}[1-9]|[0-9]{2}[1-9][0-9]{1}|[0-9]{1}[1-9][0-9]{2}|[1-9][0-9]{3})-(((0[13578]|1[02])-(0[1-9]|[12][0-9]|3[01]))|((0[469]|11)-(0[1-9]|[12][0-9]|30))|(02-(0[1-9]|[1][0-9]|2[0-8]))))|((([0-9]{2})(0[48]|[2468][048]|[13579][26])|((0[48]|[2468][048]|[3579][26])00))-02-29))\\\\s+([0-1]?[0-9]|2[0-3]):([0-5][0-9]):([0-5][0-9])"

func KafkaUI(win fyne.Window) fyne.CanvasObject {
	//标题
	title := widget.NewLabel("kafka数据导出工具")
	title.Alignment = fyne.TextAlignCenter

	//top
	mainTitle := widget.NewLabel("连接信息")
	mainTitle.Alignment = fyne.TextAlignCenter
	conEn := widget.NewEntryWithData(addrBind)
	conEn.SetPlaceHolder("例如:127.0.0.1:9092")
	addBtn := widget.NewButton("连接", func() {
		addr, _ := addrBind.Get()
		topics(&addr, win)
	})
	addBtn.Importance = widget.HighImportance
	infoBind.Set(msgWait)
	info := widget.NewLabelWithData(infoBind)
	item := widget.NewFormItem("地址", container.NewGridWithColumns(3, conEn, container.NewGridWithColumns(2, addBtn), info))
	addrForm := widget.NewForm(item)

	//left-topics
	list := widget.NewListWithData(topicsData,
		func() fyne.CanvasObject {
			return widget.NewLabel("template")
		},
		func(i binding.DataItem, o fyne.CanvasObject) {
			o.(*widget.Label).Bind(i.(binding.String))
		})
	list.OnSelected = func(id widget.ListItemID) {
		fmt.Println(id)
		topicSelected = id
	}
	scroll := container.NewVScroll(list)
	//add := widget.NewButton("Append", func() {
	//	val := fmt.Sprintf("Item %d", topicsData.Length()+1)
	//	topicsData.Append(val)
	//})

	//main-split
	//main-split-top
	start := widget.NewLabel("开始时间")
	end := widget.NewLabel("结束时间")
	start.Alignment = fyne.TextAlignCenter
	end.Alignment = fyne.TextAlignCenter
	timeEnTip := "格式:YYYY-MM-DD HH:MM:SS"
	startEn := widget.NewEntryWithData(startBind)
	endEn := widget.NewEntryWithData(endBind)
	startEn.SetPlaceHolder(timeEnTip)
	endEn.SetPlaceHolder(timeEnTip)
	row1 := container.NewGridWithColumns(2, start, end)
	row2 := container.NewGridWithColumns(2, startEn, endEn)
	var saveButton *widget.Button
	saveButton = widget.NewButton("保存数据", func() {
		addBtn.Disable()
		saveButton.Disable()
		excute(win)
		addBtn.Enable()
		saveButton.Enable()
	})
	saveButton.Importance = widget.HighImportance
	//main-split-bottom
	oneRow := widget.NewLabelWithData(oneBind)
	twoRow := widget.NewLabelWithData(twoBind)
	threeRow := widget.NewLabelWithData(threeBind)
	fourRow := widget.NewProgressBarWithData(fourBind)
	fiveRow := widget.NewLabelWithData(fiveBind)

	processForm := widget.NewForm(
		widget.NewFormItem("1 选择主题:", oneRow),
		widget.NewFormItem("2 时间范围:", twoRow),
		widget.NewFormItem("3 时间差值:", threeRow),
		widget.NewFormItem("4 开始拉取:", fourRow),
		widget.NewFormItem("5 拉取结果:", fiveRow),
	)
	rd := widget.NewRadioGroup([]string{"数据类型-不带时间戳", "数据类型-带时间戳"}, func(val string) {
		fmt.Println("radio set to ", val)
		if val == "数据类型-不带时间戳" {
			dataTime = false
		}
		if val == "数据类型-带时间戳" {
			dataTime = true
		}
	})
	rd.Selected = "数据类型-不带时间戳"
	rd.Horizontal = true
	//时间标题  时间输入框  单选按钮  确认按钮
	box := container.NewVBox(row1, row2, rd, saveButton)
	//main
	main := container.NewVSplit(box, processForm)
	main.Offset = 0.3
	split := container.NewHSplit(scroll, main)
	split.Offset = 0.2
	border := container.NewBorder(
		container.NewVBox(title, widget.NewSeparator(), addrForm, widget.NewSeparator()),
		nil,
		nil,
		nil,
		split,
	)
	initKaUiData()
	return border
}

//预置数据
func initKaUiData() {
	t := time.Now()
	t = t.Add(-time.Minute * 10)
	startTimeString := t.Format("2006-01-02 15:04:05")
	startBind.Set(startTimeString)
	t2 := time.Now()
	endTimeString := t2.Format("2006-01-02 15:04:05")
	endBind.Set(endTimeString)
	addrBind.Set("106.120.201.126:19359")
	oneBind.Set("---")
	twoBind.Set("---")
	threeBind.Set("---")
	fourBind.Set(0.0)
	fiveBind.Set("---")
}

//var cstZone = time.FixedZone("CST", 8*3600)

// 定义一个结构体来存储 JSON 数据对应的字段
type KafKaData struct {
	KafkaTime string                 `json:"KafkaTime"`
	KafkaData map[string]interface{} `json:"KafkaData"`
}

func excute(win fyne.Window) {
	//选择topic
	if topicSelected < 0 {
		err := errors.New("请先选择一个topic")
		dialog.ShowError(err, win)
		return
	}
	item, _ := topicsData.GetItem(topicSelected)
	topicName, _ := item.(binding.String).Get()
	oneBind.Set(topicName)

	//输入起止时间
	start, _ := startBind.Get()
	end, _ := endBind.Get()
	if start == "" || end == "" {
		err := errors.New("起止时间必填:格式:YYYY-MM-DD HH:MM:SS")
		dialog.ShowError(err, win)
		return
	}
	var startTime, sErr = time.ParseInLocation("2006-01-02 15:04:05", start, time.Local)
	if sErr != nil {
		err := errors.New("起止时间必填:格式:YYYY-MM-DD HH:MM:SS")
		dialog.ShowError(err, win)
		return
	}
	var endTime, eErr = time.ParseInLocation("2006-01-02 15:04:05", end, time.Local)
	if eErr != nil {
		err := errors.New("起止时间必填:格式:YYYY-MM-DD HH:MM:SS")
		dialog.ShowError(err, win)
		return
	}
	twoBind.Set(start + " 到 " + end)
	variance := endTime.Sub(startTime)
	seconds := variance.Seconds()
	i := f2i(seconds)
	threeBind.Set(FormatDuration(i))
	fmt.Println("输入正确 ", topicName, "  ", startTime, " ", endTime)

	//absolutePate := getCurrentAbPathByExecutable()
	absolutePate, _ := os.Getwd()
	fmt.Println("路径 = ", absolutePate)
	//拉数据
	serverIp, _ := addrBind.Get()
	//创建文件
	var dir = absolutePate + "\\export"
	var filePath = dir + "\\" + topicName + startTime.Format("20060102-150405") + "_" + endTime.Format("20060102-150405") + ".json"
	fmt.Println("路径 = ", filePath)
	//创建目录
	os.MkdirAll(dir, os.ModePerm)
	//删除文件
	os.Remove(filePath)
	//创建文件
	file, fErr := os.Create(filePath)
	defer file.Close()
	if fErr != nil {
		log.Fatalln(fErr)
		fiveBind.Set("失败" + fErr.Error())
		return
	}
	//
	reader := getKafkaReader(serverIp, topicName)
	defer reader.Close()
	reader.SetOffsetAt(context.Background(), startTime)
	fmt.Println("start consuming ... !!" + strconv.Itoa(int(reader.Offset())))
	if reader.Offset() < 0 {
		fmt.Println("没有数据")
		fiveBind.Set("失败，此时间段内没有数据")
		return
	}
	dataRows := 0
	for {
		//消息读取
		message, err := reader.FetchMessage(context.Background())
		if err != nil {
			log.Fatalln(err)
			fiveBind.Set("失败" + err.Error())
			break
		}

		loc, err := time.LoadLocation("Asia/Shanghai")
		if err != nil {
			fmt.Println("无法加载时区:", err)
			return
		}

		offSetTime := message.Time.In(loc)
		reader.CommitMessages(context.Background(), message)
		sub := offSetTime.Sub(startTime)
		variance.Seconds()
		gap := sub.Seconds()
		rate := gap / seconds
		rate, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", rate), 64)
		fourBind.Set(rate)
		if offSetTime.After(endTime) {
			if dataRows <= 0 {
				fmt.Println("没有数据")
				fiveBind.Set("失败，此时间段内没有数据")
				return
			}
			log.Println("end")
			fiveBind.Set("成功:" + filePath)
			fourBind.Set(1.00)
			break
		}

		var finalData = string(message.Value)
		if dataTime {
			// 创建一个空的 map[string]interface{} 来存储解析后的数据
			var data map[string]interface{}
			// 使用 Unmarshal 函数进行反序列化
			json.Unmarshal([]byte(finalData), &data)

			var kData = KafKaData{}
			kData.KafkaData = data
			kData.KafkaTime = offSetTime.Format("2006-01-02 15:04:05.000")

			var temp interface{}
			temp = kData
			jsonData, err2 := json.Marshal(temp)
			if err2 != nil {
				fmt.Println("Error encoding JSON:", err2)
				err4 := errors.New("序列化json出错")
				dialog.ShowError(err4, win)
				return
			}
			finalData = string(jsonData)
		}

		size, err := io.WriteString(file, finalData+"\r\n")
		//fmt.Printf("message at topic:%v partition:%v offset:%v	%s = %s\n", message.Topic, message.Partition, message.Offset, string(message.Key), string(message.Value))
		fmt.Println(size)
		dataRows++
		if err != nil {
			log.Println("写入出错")
			fiveBind.Set("写入文件出错" + err.Error())
			break
		}
		if message.Offset+1 == message.HighWaterMark {
			log.Println("已到最后一条")
			fiveBind.Set("成功:" + filePath)
			fourBind.Set(1.00)
			break
		}
	}
}

// 获取当前执行程序所在的绝对路径
func getCurrentAbPathByExecutable() string {
	exePath, err := os.Executable()
	if err != nil {
		log.Fatal(err)
	}
	res, _ := filepath.EvalSymlinks(filepath.Dir(exePath))
	return res
}

func getKafkaReader(kafkaURL, topic string) *kafka.Reader {
	brokers := strings.Split(kafkaURL, ",")
	return kafka.NewReader(kafka.ReaderConfig{
		Brokers: brokers,
		//GroupID:     groupID,
		Topic:       topic,
		Partition:   0,
		MaxWait:     time.Second,
		StartOffset: kafka.FirstOffset,
		//MinBytes: 1, // 10KB
		//MaxBytes: 1, // 10MB
	})
}

//查询topic列表
func topics(addr *string, win fyne.Window) {
	conn, err := kafka.Dial("tcp", *addr)
	if err != nil {
		panic(err.Error())
	}
	defer conn.Close()

	partitions, err := conn.ReadPartitions()
	if err != nil {
		errMsg := "客户端连接失败:" + err.Error()
		fmt.Printf(errMsg)
		err := errors.New(errMsg)
		dialog.ShowError(err, win)

		infoBind.Set(msgFail)
		return
	}

	topicStr := make([]string, 0)
	for _, value := range partitions {
		top := value.Topic
		b := top == "__consumer_offsets"
		if !b {
			topicStr = append(topicStr, top)
		}
	}

	fmt.Printf("topics(%d):\n", len(topicStr))
	infoBind.Set(msgSuccess)
	topicsData.Set(nil)
	sort.Strings(topicStr)
	topicsData.Set(topicStr)
}
func f2i(f float64) int {
	i, _ := strconv.Atoi(fmt.Sprintf("%1.0f", f))
	return i
}

const (
	ONE_MINTUE = 60
	ONE_HOUR   = 3600
)

func FormatDuration(duration int) string {
	if duration <= 0 {
		return "00:00:00"
	}

	if duration <= ONE_HOUR {
		m := math.Floor(float64(duration / ONE_MINTUE))
		s := duration % ONE_MINTUE
		return fmt.Sprintf("00:%02d:%02d", int(m), int(s))
	}

	h := math.Floor(float64(duration / ONE_HOUR))
	ss := math.Floor(float64(duration - int(h)*ONE_HOUR))
	var m int
	if ss >= 60 {
		m = int(math.Floor(float64(ss / ONE_MINTUE)))
	} else {
		m = 0
	}
	s := duration - (int(h*ONE_HOUR) + m*ONE_MINTUE)
	return fmt.Sprintf("%02d:%02d:%02d", int(h), int(m), s)
}
